package ithink.link.stack;
import com.alibaba.fastjson.JSONObject;
import com.aliyun.openservices.ons.api.*;
import ithink.link.ConfigHelper;
import ithink.link.base.message.LinkMessage;
import ithink.link.service.LinkService;
import ithink.link.stack.IConnection;
import ithink.link.stack.IConnectionListener;
import org.slf4j.LoggerFactory;

import java.util.Properties;
import java.util.logging.Logger;

/**
 * Created by zaqb on 17-12-1.
 */


public class MqConnection implements IConnection{


    LinkStack.IListener listener;
    private Consumer consumer;
    private Producer producer;
    private Object context;
    private static final org.slf4j.Logger LOGGER = LoggerFactory.getLogger(MqConnection.class);

    public MqConnection( LinkStack.IListener listener) {
        initConsumer(listener);
        initProducer();
    }


    private void initConsumer( LinkStack.IListener listener) {
        Properties properties = new Properties();
        properties.put(PropertyKeyConst.ConsumerId, ConfigHelper.getMqConsumerId());
        properties.put(PropertyKeyConst.AccessKey, ConfigHelper.getMqAccessKey());
        properties.put(PropertyKeyConst.SecretKey, ConfigHelper.getMqAccessSecret());
        properties.put(PropertyKeyConst.ONSAddr, ConfigHelper.getMqAddr());
        consumer = ONSFactory.createConsumer(properties);

        this.listener = listener;

        final MqConnection mqConnection = this;
        consumer.subscribe(ConfigHelper.getMqTopic(), ConfigHelper.getName(), new MessageListener() {

            public Action consume(Message message, ConsumeContext context) {
                try {
                    LOGGER.info( "recv mes from mq :" + message);
                    JSONObject jsonObject = JSONObject.parseObject( new String( message.getBody() ) );
                    LinkMessage message1 = new LinkMessage();
                    message1.setFlag( LinkMessage.FLAG_PART_KEY );
                    message1.setText( jsonObject );
                    mqConnection.listener.onMessageReceived( mqConnection, message1 );
                }catch (Exception ex) {
                    LOGGER.error( "consume ex :{}", ex );
                }
                return Action.CommitMessage;
            }
        });
        LOGGER.info("consumer, ConsumerId = {}", ConfigHelper.getMqConsumerId());
        LOGGER.info("consumer, AccessKey = {}", ConfigHelper.getMqAccessKey());
        LOGGER.info("consumer, SecretKey = {}", ConfigHelper.getMqAccessSecret());
        LOGGER.info("consumer, ONSAddr = {}", ConfigHelper.getMqAddr());

        LOGGER.info("consumer, topic = {}", ConfigHelper.getMqTopic());
        LOGGER.info("consumer, tag = {}", ConfigHelper.getName());

        consumer.start();
    }

    private void initProducer() {
        Properties properties = new Properties();
        properties.put(PropertyKeyConst.ProducerId, ConfigHelper.getMqProducerId());
        properties.put(PropertyKeyConst.AccessKey, ConfigHelper.getMqAccessKey());
        properties.put(PropertyKeyConst.SecretKey, ConfigHelper.getMqAccessSecret());
        properties.put(PropertyKeyConst.ONSAddr, ConfigHelper.getMqAddr());
        properties.setProperty(PropertyKeyConst.SendMsgTimeoutMillis, "3000");
        producer = ONSFactory.createProducer(properties);


        LOGGER.info("producer, ConsumerId = {}", ConfigHelper.getMqProducerId());
        LOGGER.info("producer, AccessKey = {}", ConfigHelper.getMqAccessKey());
        LOGGER.info("producer, SecretKey = {}", ConfigHelper.getMqAccessSecret());
        LOGGER.info("producer, ONSAddr = {}", ConfigHelper.getMqAddr());
        producer.start();

        LOGGER.info("producer, topic = {}", ConfigHelper.getMqTopic());
    }
    @Override
    public State getState() {
        return null;
    }

    @Override
    public String remoteIp() {
        return "";
    }

    @Override
    public int remotePort() {
        return 0;
    }

    @Override
    public Object getContext() {
        return context;
    }

    @Override
    public void setContext(Object context) {
        this.context = context;
    }

    @Override
    public void sendMessage(LinkMessage message) {
        message.getText().put( "sender", ConfigHelper.getName() );
        String tag;
        if(!message.getText().containsKey( "tag" ) || message.getText().getString( "tag" ) == null) {
            LOGGER.warn( "send mq not contains tag, use default" );
            tag = "web1";
        }else {
            tag = message.getText().getString( "tag" );
        }
        Message msg = new Message(
                ConfigHelper.getMqTopic(),
                tag,
                message.getText().toJSONString().getBytes());
        LOGGER.info( "send msg to mq: tag: {}, msg:{}", tag, message.getText().toJSONString() );
        producer.sendAsync(msg, new SendCallback() {
            @Override
            public void onSuccess(final SendResult sendResult) {
            }

            @Override
            public void onException(OnExceptionContext context) {
                LOGGER.error( "send message error, id = {}, ex{}",
                        context.getMessageId(), context.getException() );

            }
        });
    }

    @Override
    public void disconnect() {

    }

    @Override
    public void disconnect(Exception ex) {

    }

    @Override
    public void setListener(IConnectionListener listener) {

    }


    public static void main(String[] args) {
    }
}